package com.jy.oms.framework.asyn.task.core;

import cn.hutool.core.thread.ThreadUtil;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessage;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessageFacade;
import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.thread.CloseableThread;
import com.jy.oms.framework.util.DateUtils;
import com.jy.oms.framework.util.NetworkUtil;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * WorkerServer心跳线程管理器，保证只有一台机器工作
 *
 * @author liuquan
 */
public class WorkerServerThreadManager {
    private static WorkerServerThreadManager INSTANCE = new WorkerServerThreadManager();
    private volatile boolean IS_WORKER_SERVER = false;
    private AtomicBoolean INIT_THREAD = new AtomicBoolean();
    private AtomicInteger COUNT = new AtomicInteger();

    private WorkerServerThreadManager() {
    }

    public static WorkerServerThreadManager getInstance() {
        return INSTANCE;
    }

    /**
     * 启动心跳线程
     */
    public void startWorkerServerThread() {
        if (INIT_THREAD.compareAndSet(false, true)) {
            Thread workerServerThread = new WorkerServerThread();
            workerServerThread.setDaemon(true);
            workerServerThread.setName("WorkerServerThread");
            workerServerThread.start();
        }
    }

    public boolean isWorkerServer() {
        return IS_WORKER_SERVER;
    }

    private class WorkerServerThread extends CloseableThread {

        @Override
        public void runCloseable() {
            this.check();

            ThreadUtil.sleep(60, TimeUnit.SECONDS);
        }

        private void check() {
            RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
            String workerServerKey = CacheConstants.genCacheKey(CacheConstants.WORKER_SERVER_RUN);
            RBucket<String> jobServerBucket = redissonClient.getBucket(workerServerKey);
            String jobServerIp = jobServerBucket.get();
            if (StringUtils.isBlank(jobServerIp)) {
                String workerServerJudgeKey = CacheConstants.genCacheKey(CacheConstants.WORKER_SERVER_RUN_JUDGE);
                RBucket<Object> judgeJobRunBucket = redissonClient.getBucket(workerServerJudgeKey);
                boolean judgeResult = judgeJobRunBucket.compareAndSet(null, true);
                if (judgeResult) {
                    jobServerIp = NetworkUtil.getLocalhostStr();
                    //运行的JOB可能停止了，修改IP为本JOB的IP，进行运行
                    jobServerBucket.set(jobServerIp, 90, TimeUnit.SECONDS);

                    IS_WORKER_SERVER = true;

                    int count = COUNT.getAndAdd(1);
                    if(count > 1){
                        AlarmMessage alarmMessage = AlarmMessage.create()
                                .title("切换WorkerServer")
                                .add("IP", jobServerIp)
                                .add("时间", DateUtils.date2String(new Date(), DateUtils.YYYY_MM_DD_HH_MM_SS));
                        AlarmMessageFacade.getInstance().alarm(alarmMessage);
                    }
                } else {
                    IS_WORKER_SERVER = false;
                }

                //保证异常情况下(重启时redis断开,可能不会设置失效时间)，judge缓存不会永远是有效状态
                judgeJobRunBucket.expire(30, TimeUnit.SECONDS);
            } else {
                boolean isRun = NetworkUtil.getLocalhostStr().equals(jobServerIp);
                if (isRun) {
                    //能执行的ip，才能给续命
                    jobServerBucket.set(jobServerIp, 90, TimeUnit.SECONDS);

                    IS_WORKER_SERVER = true;
                } else {
                    IS_WORKER_SERVER = false;
                }
            }
        }

    }
}
